package day06

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.8	RDD持久化
 */
object Persistence {
}

/**
 * 2.8.1	RDD Cache缓存
 *
 * RDD通过Cache或者Persist方法将前面的计算结果缓存，默认情况下会把数据
 * 以序列化的形式缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，
 * 而是触发后面的action，该RDD才会被缓存在计算节点的内存中，并供后面重用。
 *
 * cache操作会增加血缘关系，不改变原有的血缘关系
 *
 * 1、缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，RDD的缓存容错机制
 * 保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换，丢失的
 * 数据会被重算，由于RDD的各个Partition是相对独立的，因此只需要计算丢失的
 * 部分即可，并不需要重算全部Partition。
 *
 * 2、自带缓存算子
 * Spark会自动对一些Shuffle操作的中间数据做持久化操作(比如：reduceByKey)。
 * 这样做的目的是为了当一个节点Shuffle失败了避免重新计算整个输入。但是，在
 * 实际使用的时候，如果想重用数据，仍然建议调用persist或cache。
 */
object Persistence_cache {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val lineRdd: RDD[String] = sc.textFile("input")

    val wordRdd: RDD[String] = lineRdd.flatMap(_.split(" "))

    val tupleRdd: RDD[(String, Long)] = wordRdd.map {
      word => {
        println("------执行算子时打印的内容，如果是从缓存中读取，此日志不会打印------")
        (word, System.currentTimeMillis())
      }
    }

    println("-----原有的血缘关系-----")
    println(tupleRdd.toDebugString)

    println("--------调用cache之前触发一次执行逻辑--------")
    tupleRdd.collect().foreach(println)

    println("-------调用cache进行缓存，将wordRdd调用的map算子计算的结果缓存起来---------")
    tupleRdd.cache()
    // cache()底层调用的是persist()，默认的存储级别：StorageLevel.MEMORY_ONLY，仅在内存存储一份
    // 可以更改存储级别
    // tupleRdd.persist(StorageLevel.MEMORY_AND_DISK_2)

    println("------调用cache之后的血缘关系-------")
    println(tupleRdd.toDebugString)

    // 触发执行逻辑
    println("---------触发行动算子-------------")
    tupleRdd.collect().foreach(println)

    // 再次触发执行逻辑
    println("-------再次触发行动算子-----------")
    tupleRdd.collect().foreach(println)

    // cache操作会增加血缘关系，不改变原有的血缘关系

    // 使用完缓存，调用unpersist()方法释放缓存
    tupleRdd.unpersist()

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.8.2	RDD CheckPoint检查点
 *
 * 1）检查点：是通过将RDD中间结果写入磁盘。
 *
 * 2）为什么要做检查点？
 * 由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，
 * 如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。
 *
 * 3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统
 *
 * 4）检查点数据存储格式为：二进制的文件
 *
 * 5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。
 *
 * 6）检查点触发时间：对RDD进行checkpoint操作并不会马上被执行，必须执行Action
 * 操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍
 *
 * 7）设置检查点步骤
 * （1）设置检查点数据存储路径：sc.setCheckpointDir("./checkpoint1")
 * （2）调用检查点方法：wordToOneRdd.checkpoint()
 *
 * 案例：只增加checkpoint，没有增加Cache缓存打印
 */
object Persistence_checkPoint {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.设置检查点存储路径。如果在群集上运行，该目录必须是HDFS路径。
    sc.setCheckpointDir("./checkpoint1")

    val lineRdd: RDD[String] = sc.textFile("input")

    val wordRdd: RDD[String] = lineRdd.flatMap(_.split(" "))

    val tupleRdd: RDD[(String, Long)] = wordRdd.map {
      println("------------执行算子时打印的内容，如果是从检查点读取，此日志不会打印---------------")
      (_, System.currentTimeMillis())
    }

    println("------设置检查点之前，打印血缘关系-------")
    println(tupleRdd.toDebugString)

    //4.设置数据检查点
    tupleRdd.checkpoint()

    println("------设置检查点之后，打印血缘关系，发现检查点还没有切断血缘。")
    println(tupleRdd.toDebugString)

    println("------触发执行逻辑，会对RDD进行checkpoint操作------")
    tupleRdd.collect().foreach(println)
    println("------触发执行逻辑之后，打印血缘关系，发现检查点切断血缘，该RDD的所有依赖于父RDD中的信息全部被移除。")
    println(tupleRdd.toDebugString)

    // 第1个job执行完，触发了checkpoint，第2个job运行checkpoint，并把数据存储在检查点上0。
    // 第3、4个job，数据从检查点上直接读取。

    println("------再次触发执行逻辑，检查点为了数据安全，会从血缘关系的最开始执行一遍，不会打印执行算子中的日志------")
    tupleRdd.collect().foreach(println)
    println("----------------------------------------------------------------")
    println("------第三次触发执行逻辑，从检查点中取数据，也不会打印执行算子中的日志了------")
    tupleRdd.collect().foreach(println)
    println("----------------------------------------------------------------")

    // 关闭连接
    sc.stop()
  }
}

/**
 * 实际上检查点会结合cache一起使用，这样避免了做checkpoint时重新计算
 *
 * 案例：增加checkpoint，也增加Cache缓存打印
 */
object Persistence_checkPointNew {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.设置检查点
    sc.setCheckpointDir("./checkpoint2")

    val lineRdd: RDD[String] = sc.textFile("input")

    val wordRdd: RDD[String] = lineRdd.flatMap(_.split(" "))

    val tupleRdd: RDD[(String, Long)] = wordRdd.map((_, System.currentTimeMillis()))

    //4.增加缓存，避免再重跑一个job做checkpoint
    println("---------加缓存前，打印血缘关系---------")
    println(tupleRdd.toDebugString)
    tupleRdd.cache()
    println("---------加缓存后，打印血缘关系---------")
    println(tupleRdd.toDebugString)

    println("---------标记检查点之前，打印血缘关系---------")
    println(tupleRdd.toDebugString)
    //5.数据检查点，针对tupleRdd做检查点计算
    tupleRdd.checkpoint()
    println("---------标记检查点之后，打印血缘关系---------")
    println(tupleRdd.toDebugString)

    println("连续三次触发行动操作，发现三次的结果完全一样，这是预期的结果")
    tupleRdd.collect().foreach(println)

    // 第1个job执行完，数据就保存到Cache里面了，第2个job运行checkpoint，直接读取Cache里面的数据，
    // 并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。

    println("---------标记检查点之后并触发行动操作之后，打印血缘关系---------")
    println(tupleRdd.toDebugString)
    println("------------------------")
    tupleRdd.collect().foreach(println)
    println("------------------------")
    tupleRdd.collect().foreach(println)

    //6.线程休眠，访问http://localhost:4040/jobs/，查看DAG图
    Thread.sleep(1000000)

    // 使用完缓存，通过unpersist()释放缓存
    tupleRdd.unpersist()

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.8.3	缓存和检查点区别
 *
 * 1）Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
 *
 * 2）Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等
 * 容错、高可用的文件系统，可靠性高。
 *
 * 3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据
 * 即可，否则需要再从头计算一次RDD。
 *
 * 4）如果使用完了缓存，可以通过unpersist()方法释放缓存
 *
 */
object Persistence_cache_checkPoint {

}

/**
 * 2.8.4	检查点存储到HDFS集群
 *
 * 如果检查点数据存储到HDFS集群，要注意配置访问集群的用户名。否则会报访问权限异常。
 */
object Persistence_checkPoint02 {
  def main(args: Array[String]): Unit = {

    // 设置访问HDFS的用户名，否则会报org.apache.hadoop.security.AccessControlException: Permission denied:
    System.setProperty("HADOOP_USER_NAME", "atguigu")

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.设置检查点存储路径，需要提前在HDFS集群上创建/checkpoint路径(亲测：hadoop3.1.3不需要提前创建)
    sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")

    val lineRdd: RDD[String] = sc.textFile("input")

    val tupleRdd: RDD[(String, Long)] = lineRdd.flatMap(_.split(" ")).map((_, System.currentTimeMillis()))

    //4.增加缓存，避免再重新跑一个job做checkpoint
    tupleRdd.cache()

    //5.数据检查点：针对tupleRdd做检查点计算
    tupleRdd.checkpoint()

    tupleRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}
